package com.us.kafka;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;

import java.math.BigDecimal;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * 
 * @author yangyibo
 *
 */
public class KafkaProducerDriverBase extends Thread{
	 private String topic;
	 private JSONObject jsonObject;

    private JSONArray jsonArray;

	    public KafkaProducerDriverBase(String topic, JSONObject jsonObject){
	        super();  
	        this.topic = topic;
	        this.jsonObject = jsonObject;
	    }
    public KafkaProducerDriverBase(String topic, JSONArray jsonArray){
        super();
        this.topic = topic;
        this.jsonArray = jsonArray;
    }
	      
	    @Override  
	    public void run() {  
	        Producer producer = createProducer();  
	        int i=1;
	        while(i<=1){
	        	i++;
                System.out.println("jsonArray  =    " + jsonArray.toString());
	            producer.send(new KeyedMessage<Integer, String>(topic, jsonArray.toString()));
	            try {
	                TimeUnit.SECONDS.sleep(1);
	            } catch (InterruptedException e) {  
	                e.printStackTrace();  
	            }  
	        }  
	    }  
	  
		private Producer createProducer() {  
	        Properties properties = new Properties();
			//声明zk
	        properties.put("zookeeper.connect", KafkaConfig.zookeeper);
	        properties.put("serializer.class", StringEncoder.class.getName());
			// 声明kafka broker
	        properties.put("metadata.broker.list", KafkaConfig.metadata_broker_list);
            properties.put("acks", "all");
            properties.put("retries", 0);
            properties.put("batch.size", 16384);
            properties.put("linger.ms", 1);
            properties.put("buffer.memory", 33554432);
            properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            return new Producer<Integer, String>(new ProducerConfig(properties));
	     }  
	      
	      
		/**
		 * 
		 * @param topicName
		 */
	    public static void creatProducer(String topicName ,JSONArray json) {

            JSONArray jsonArray = new JSONArray();
            for(int i = 0;i<10000;i++){
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("deviceCatalog","车辆-car");
                jsonObject.put("deviceMark","鄂A"+i+i+i+i);
                jsonObject.put("deviceType","15");
                jsonObject.put("stationId","420100000000");
                jsonObject.put("updateTime",DateUtil.getToDate("yyyy-MM-dd HH:mm:ss"));
                jsonArray.add(jsonObject);
            }
	        new KafkaProducerDriverBase(topicName,jsonArray).start();
	    }  
	       
	} 